package app.dws;

import app.BaseAppV1;
import beans.TradeCourseOrderBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import function.DimAsyncFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import utils.AtguiguUtil;
import utils.FlinkSinkUtil;

import java.time.Duration;
import java.util.Collections;
import java.util.concurrent.TimeUnit;

public class Dws_0301_DwsTradeCourseOrderWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_0301_DwsTradeCourseOrderWindow().init(3031, 2,
                "DwsTradeCourseOrderWindow",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }

    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //维度之前
        SingleOutputStreamOperator<TradeCourseOrderBean> aggregatedStreamWithoutDims = aggregatedStreamWithoutDims(stream);
        //维度之后
        SingleOutputStreamOperator<TradeCourseOrderBean> resultStream = addDims(aggregatedStreamWithoutDims);
        //写入clickHouse
        writeToClickHouse(resultStream);
    }

    private void writeToClickHouse(SingleOutputStreamOperator<TradeCourseOrderBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_course_order_window", TradeCourseOrderBean.class));
    }

    private SingleOutputStreamOperator<TradeCourseOrderBean> addDims(SingleOutputStreamOperator<TradeCourseOrderBean> aggregatedStreamWithoutDims) {
        SingleOutputStreamOperator<TradeCourseOrderBean> courseIdStream = AsyncDataStream.unorderedWait(
                aggregatedStreamWithoutDims,
                new DimAsyncFunction<TradeCourseOrderBean>() {
                    public String table() {
                        return "dim_course_info";
                    }

                    public String id(TradeCourseOrderBean input) {
                        return input.getCourseId();
                    }

                    public void addDim(JSONObject dim, TradeCourseOrderBean input) {
                        input.setCourseName(dim.getString("COURSE_NAME"));
                        input.setSubjectId(dim.getString("ID"));
                    }
                }, 120L, TimeUnit.SECONDS);
        SingleOutputStreamOperator<TradeCourseOrderBean> subjectIdStream = AsyncDataStream.unorderedWait(
                courseIdStream, new DimAsyncFunction<TradeCourseOrderBean>() {
                    public String table() {
                        return "dim_subject_info";
                    }

                    public String id(TradeCourseOrderBean input) {
                        return input.getSubjectId();
                    }

                    public void addDim(JSONObject dim, TradeCourseOrderBean input) {
                        input.setSubjectName(dim.getString("SUBJECT_NAME"));
                        input.setCategoryId(dim.getString("ID"));
                    }
                }, 120L, TimeUnit.SECONDS);
        return AsyncDataStream.unorderedWait(
                subjectIdStream,
                new DimAsyncFunction<TradeCourseOrderBean>() {
                    public String table() {
                        return "dim_category_info";
                    }

                    public String id(TradeCourseOrderBean input) {
                        return input.getCategoryId();
                    }

                    public void addDim(JSONObject dim, TradeCourseOrderBean input) {
                        input.setCategoryName(dim.getString("CATEGORY_NAME"));
                    }
                }, 120L, TimeUnit.SECONDS);
    }

    private SingleOutputStreamOperator<TradeCourseOrderBean> aggregatedStreamWithoutDims(DataStreamSource<String> stream) {
        return stream
                .map(JSON::parseObject)
                //pojo
                .map((obj) -> {
                    return TradeCourseOrderBean.builder()
                            .courseId(obj.getString("course_id"))
                            .courseName(obj.getString("course_name"))
                            .userIdSet(Collections.singleton(obj.getString("user_id")))
                            .userCount(0L)
                            .orderIdSet(Collections.singleton(obj.getString("order_id")))
                            .orderCount(0L)
                            .finalAmount(obj.getBigDecimal("final_amount"))
                            .ts(obj.getLong("ts") * 1000L)
                            .build();
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TradeCourseOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                        .withTimestampAssigner((bean, ts) -> {
                            return bean.getTs();
                        }))
                .keyBy(TradeCourseOrderBean::getCourseId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5L)))
                .reduce(new ReduceFunction<TradeCourseOrderBean>() {
                            public TradeCourseOrderBean reduce(TradeCourseOrderBean value1, TradeCourseOrderBean value2) throws Exception {
                                value1.getUserIdSet().addAll(value2.getUserIdSet());
                                value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                                value1.setFinalAmount(value1.getFinalAmount().add(value2.getFinalAmount()));
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<TradeCourseOrderBean, TradeCourseOrderBean, String, TimeWindow>() {
                            public void process(String s,
                                                Context context,
                                                Iterable<TradeCourseOrderBean> elements,
                                                Collector<TradeCourseOrderBean> out) throws Exception {
                                TradeCourseOrderBean bean = (TradeCourseOrderBean) elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(((TimeWindow) context.window()).getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(((TimeWindow) context.window()).getEnd()));

                                bean.setUserCount((long) bean.getUserIdSet().size());
                                bean.setOrderCount((long) bean.getOrderIdSet().size());

                                bean.setTs(System.currentTimeMillis());
                                out.collect(bean);
                            }
                        });
    }
}